package com.shujia.filter;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class Code07ScoreFilter {

    // 将所有学生总分数据大于450分学生过滤出来
    Connection connection;
    @Before
    public void getConnection(){
        Configuration configuration = new Configuration();

        // 注意：需要在当前平台中配置node1 node2的主机映射
        configuration.set("hbase.zookeeper.quorum","192.168.163.100,node1,node2");
        try {
            connection  = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void findScoreMore450() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));
        Scan scan = new Scan();
        FamilyFilter score = new FamilyFilter(
                CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("score")
        );
        scan.setFilter(score);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {

//            ResultUtil.printRes(result);
            String rowKey = Bytes.toString(result.getRow());
            int totalScore = 0;
            for (Cell cell : result.listCells()) {
                // 将每个列中的Value成绩取出
                String scoreStr = Bytes.toString(CellUtil.cloneValue(cell));
                totalScore += Integer.parseInt(scoreStr);
            }

            // 如果总分大于450分，那么可以再对表中的数据进行过滤
            if (totalScore > 450){
                // 对于单个的rowKey再进行过滤操作
                Result oneRowKeyRes = table.get(new Get(rowKey.getBytes()));
                System.out.print(totalScore+"\t");
                ResultUtil.printRes(oneRowKeyRes);
            }

        }
    }


    @Test
    public void addTotalScore() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));
        Scan scan = new Scan();
        FamilyFilter score = new FamilyFilter(
                CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("score")
        );
        scan.setFilter(score);

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());
            int totalScore = 0;
            for (Cell cell : result.listCells()) {
                // 将每个列中的Value成绩取出
                String scoreStr = Bytes.toString(CellUtil.cloneValue(cell));
                totalScore += Integer.parseInt(scoreStr);
            }
            // 将计算的结果再保存到Hbase表中
            //  需要添加一个列族，专门用于保存计算结果

            // 通过Put将计算的总分数据添加到源表中
            Put put = new Put(rowKey.getBytes());
            put.addColumn("compute".getBytes(),"total_score".getBytes(),Bytes.toBytes(totalScore));
            table.put(put);
        }
    }

    @Test
    public void getScoreMore450() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "compute".getBytes(),
                "total_score".getBytes(),
                CompareOperator.GREATER,
                Bytes.toBytes(450)
        );

        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            // 对于添加数据时的数据类型要和获取数据后转换的类型一致
            ResultUtil.printRes(result,"total_score");
        }
    }

    // 对于各班大于450分的同学进行统计个数
    @Test
    public void getMore450Cnt() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "compute".getBytes(),
                "total_score".getBytes(),
                CompareOperator.GREATER,
                Bytes.toBytes(450)
        );

        scan.setFilter(singleColumnValueFilter);

        ResultScanner scanner = table.getScanner(scan);

        // 对于HashMap中并不能存储过多的数据 对于大批量数据计算，这时需要使用MR
        HashMap<String, Integer> clazzNum = new HashMap<>();

        for (Result result : scanner) {

            // 1.先判断其分数是否大于450
            int total_score = Bytes.toInt(result.getValue("compute".getBytes(),
                    "total_score".getBytes()
            ));

            if (total_score > 450){
                String clazz = Bytes.toString(result.getValue("info".getBytes(),
                        "clazz".getBytes()
                ));

                Integer cnt = clazzNum.getOrDefault(clazz,0);
                cnt += 1;

                clazzNum.put(clazz,cnt);
            }


            // 2.如果大于，那么需要获取其班级信息，再将对应班级的信息后 + 1
        }

        for (Map.Entry<String, Integer> entry : clazzNum.entrySet()) {
            System.out.println(entry);
        }
    }

    @After
    public void close(){
        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


}
